package com.hadwinling.alogriithm.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object SparkStreaming02_Queue {
  def main(args: Array[String]): Unit = {
    //todo 创建环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(conf, Seconds(3))

//    创建 RDD 队列
    var rddQueue = new mutable.Queue[RDD[Int]]()

    //创建QueueInputDStream
    val inputStream =ssc.queueStream(rddQueue,oneAtATime = false)

    //处理队列中的rdd数据
    val mappedStream = inputStream.map((_,1))
    val reduceStream = mappedStream.reduceByKey(_+_)

    //打印结果
    reduceStream.print()

    //启动任务
    ssc.start()


    //循环创建并向rdd对象中放入rdd
    for(i <- 1 to 5){
      rddQueue += ssc.sparkContext.makeRDD(1 to 300 ,10)
      Thread.sleep(2000)
    }
    // 2. 等待采集器的关闭
    ssc.awaitTermination()

  }

}
